package org.example.flumeinterceptor;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.codec.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class DateInterceptorV2 implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        Map<String, String> eventHeader = event.getHeaders();
        String body = new String(event.getBody(), Charsets.UTF_8);

        String[] splitStrings = body.split("\\s+");
        try {
            String splitString = splitStrings[6];
            JSONObject jsonObject = JSONObject.parseObject(splitString);


            String logtype = eventHeader.get("logtype");
            Long time = 0L;
            if (logtype.equals("start")) {
                time = jsonObject.getJSONObject("app_active").getLong("time");
            }
            else if (logtype.equals("event")) {
                time = jsonObject.getJSONArray("lagou_event").getJSONObject(0).getLong("time");
            }

            LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(time), ZoneId.systemDefault());
            DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            String formattedDate = localDateTime.format(dateTimeFormatter);

            eventHeader.put("time", formattedDate);
        } catch (Exception e) {
            e.printStackTrace();
            eventHeader.put("time", "unknown");
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> eventLists = new ArrayList<>();
        for (Event event : list) {
            Event outEvent = intercept(event);
            if (outEvent != null) {
                eventLists.add(outEvent);
            }

        }
        return eventLists;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new DateInterceptorV2();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
